package com.athui.utils.explorer.metadata;

import com.athui.bean.metadata.ColumnMetaData;
import com.athui.bean.config.ConnectInfo;
import com.athui.bean.config.DataType;
import com.athui.bean.metadata.TableMetaData;
import com.athui.utils.common.jdbc.JdbcUtils;
import com.athui.utils.common.jdbc.JdbcResultSetBuilder;
import org.apache.commons.codec.digest.DigestUtils;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import java.sql.*;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * @description: TODO 元数据探查
 * @autor: zhangzhonghui
 * @create: 2024-08-01 09:12
 * @Version: 1.0
 */
public abstract class JdbcExplorer extends JdbcUtils implements MetaDataExplorer{


    public  abstract Map<String, DataType> typeMapping();

    public abstract TableMetaData tableMetaDataBuilder(String schemaName, String tableName);

    public abstract ColumnMetaData columnMetaDataBuilder(ColumnMetaData columnMetaData);


    /**
     * @description: TODO 完成一下初始化操作
     * @author: 阿辉大人
     * @param: datableName
     * @param: tableName
     * @create: 2024/8/6 19:30
     * @version: 1.0
     **/
    protected void open(String schemaName,String tableName){}


    /**
     * @description: TODO 获取表的元数据信息
     * @autor: zhangzhonghui
     * @param: datableName  库名
     * @param: tableName    表名
     * @return: com.athui.bean.TableMetaData
     * @create: 2024/8/2 12:50
     * @Version: 1.0
     **/
    public  TableMetaData build(String schemaName,String tableName){
        System.out.println(" build " + schemaName+  "." + tableName);
        // 初始化
        open(schemaName,tableName);

        TableMetaData tableMetaData = scanTableMetaData(schemaName, tableName);
        tableMetaData.setSchemaName(schemaName);
        tableMetaData.setTableName(tableName);
        List<ColumnMetaData> columnMetaDataList = scanColumnMetaData(schemaName, tableName)
                .stream()
                .peek(e -> e.setTableCode(tableMetaData.getTableCode()))
                .collect(Collectors.toList());
        tableMetaData.setColumnSize(columnMetaDataList.size());
        tableMetaData.setColumnMetaDataList(columnMetaDataList);

        // 程序结束后执行
        close();
        return tableMetaData;
    }

    public List<TableMetaData> build(Map<String, String> tableMap){
        return tableMap.keySet().stream()
                .map(tbl ->build(tableMap.get(tbl), tbl))
                .collect(Collectors.toList());
    }




    /**
     * @description: TODO 统计表的元数据信息
     * @author: 阿辉大人
     * @param: schemaPattern
     * @param: tableNamePattern
     * @return: com.athui.bean.TableMetaData
     * @create: 2024/8/4 18:53
     * @version: 1.0
     **/
    protected TableMetaData scanTableMetaData(String schemaPattern,String tableNamePattern){


        TableMetaData  tableMetaData = tableMetaDataBuilder(schemaPattern,tableNamePattern);

        if(tableMetaData == null){
            tableMetaData = super.operation(new JdbcResultSetBuilder<TableMetaData>() {
                @Override
                public String buildSql() {
                    return String.format("select count(1) cou from %s.%s", schemaPattern, tableNamePattern);
                }

                @Override
                public TableMetaData accept(ResultSet rs) throws SQLException {
                    // 查看 Table 表的总数据量
                    TableMetaData obj = new TableMetaData();
                    while (rs.next()){
                        obj.setRowSize(rs.getInt(1));
                    }
                    return obj;
                }
            });
        }
        return  tableMetaData;
    }

    /**
     * @description: TODO 获取 表字典的元数据信息
     * @author: 阿辉大人
     * @param: schemaPattern 库名
     * @param: tableNamePattern 表名
     * @return: java.util.ArrayList<com.athui.bean.ColumnMetaData>
     * @create: 2024/8/3 11:36
     * @version: 1.0
     **/
    protected ArrayList<ColumnMetaData> scanColumnMetaData(String schemaPattern,String tablePattern){

        Connection connection = null;
        ResultSet primaryKeys = null;
        ResultSet tableResult = null;

        ArrayList<ColumnMetaData> columnMetaDataList = new ArrayList<>();

        // https://blog.csdn.net/m4330187/article/details/118458981
        try {
            connection = super.getConnection();

            DatabaseMetaData metaData = connection.getMetaData();

            // flink 类型映射
            Map<String, DataType> flinkTypeMapping =  typeMapping();


            // 获取表中的主键
            primaryKeys = metaData.getPrimaryKeys(null, schemaPattern, tablePattern);

            Map<String,String> primaryKeyMap =new HashMap<>();

            while (primaryKeys.next()){
                String pkName = primaryKeys.getString("PK_NAME");
                primaryKeyMap.put(primaryKeys.getString("COLUMN_NAME"),pkName);
            }

            // 表的元数据扫描
            tableResult = metaData.getColumns(null, schemaPattern,tablePattern,null);

            int index = 0;
            while (tableResult.next()){

                String dataBaseName = tableResult.getString("TABLE_SCHEM");
                // 库名
                String schemaName = tableResult.getString("TABLE_CAT");
                // 表名
                String tableName = tableResult.getString("TABLE_NAME");
                String columnName = tableResult.getString("COLUMN_NAME");

                if (!((schemaPattern.equals(dataBaseName) || schemaPattern.equals(schemaName)) && tablePattern.equals(tableName))){
                    continue;
                }

                index+=1;

                String columnCode = String.format("%s_%s_%s_%s",
                        dataBaseName,
                        schemaName,
                        tableName,
                        columnName
                );

                ColumnMetaData columnMetaData = new ColumnMetaData();
                // 编号
                columnMetaData.setColumnCode(DigestUtils.md5Hex(columnCode));
                // 序号，计算完成之后，字段顺序可能会发生改变，用于排序
                columnMetaData.setIndex(String.valueOf(index));
                // 列名
                columnMetaData.setColumnName(columnName);
                // 列类型
                columnMetaData.setOriginalColumnType(tableResult.getString("TYPE_NAME"));
                // 列描述
                columnMetaData.setColumnComment(tableResult.getString("REMARKS"));
                // 列的精度
                columnMetaData.setColumnPrecision(tableResult.getInt("COLUMN_SIZE"));
                // 列的标度
                columnMetaData.setColumnScale(tableResult.getInt("DECIMAL_DIGITS"));
                // 1 可以是 null ，0 不能为 null
                columnMetaData.setAllowNull(tableResult.getInt("NULLABLE") == 1);

                // 是否为 主键
                String keyType = primaryKeyMap.get(columnMetaData.getColumnName());
                columnMetaData.setKey( keyType != null);
                if(columnMetaData.getKey()){
                    columnMetaData.setKeyType(keyType);
                }

                // 设置 Flink 数据类型
                DataType dataType = flinkTypeMapping.get(columnMetaData.getOriginalColumnType());
                // 转换成 Int 主要是用于观察异常
                if (dataType == null){
                    dataType = new DataType(BasicTypeInfo.INT_TYPE_INFO, DataType.NUMERICAL);
                }

                // 设置 flink 数据类型
                columnMetaData.setFlinkColumnType(dataType.getFlinkType());
                // 设置 类型标签
                columnMetaData.setTypeLabel(dataType.getLabel());

                columnMetaDataList.add(columnMetaDataBuilder(columnMetaData));
            }
        }catch (SQLException | ClassNotFoundException e){
            e.printStackTrace();
        }finally {
            // 释放数据库连接
            super.close(connection,primaryKeys);
            super.close(connection,tableResult);
        }

        return columnMetaDataList;
    }

    public MetaDataExplorer setConnectInfo(ConnectInfo connectInfo) {
        super.putConnectInfo(connectInfo);
        return this;
    }


    /**
     * @description: TODO 程序结束后执行
     * @author: 阿辉大人
     * @create: 2024/8/6 19:30
     * @version: 1.0
     **/
    protected  void close(){}
}
